package com.example.lagoues;

import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.client.IndicesClient;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.CreateIndexResponse;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.client.indices.GetIndexResponse;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;
import org.elasticsearch.index.query.QueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.web.bind.annotation.*;

import java.io.IOException;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.BiConsumer;
import java.util.stream.Collectors;

@RestController
@RequestMapping("/position")
public class PositionController {
    
    private static final String INDEX_NAME = "position";
    
    @Autowired
    private JdbcTemplate jdbcTemplate;
    @Autowired
    private RestHighLevelClient restHighLevelClient;
    
    @GetMapping
    public List<Position> search(@RequestParam("positionName") String positionName) throws IOException {
       
        QueryBuilder queryBuilder = QueryBuilders.matchQuery("positionName", positionName);
        
        List<Position> positions = search(queryBuilder);
        if (positions.size() < 5) {
        
            QueryBuilder queryBuilder2 = QueryBuilders.matchQuery("positionAdvantage", "美女多 OR 员工福利好");
            List<Position> tmp = search(queryBuilder2);
        
            for (Position position : tmp) {
                if (!positions.contains(position)) {
                    positions.add(position);
                }
            }
        }
    
        return positions;
    }
    
    private List<Position> search(QueryBuilder queryBuilder) throws IOException {
        
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(queryBuilder);
        
        SearchRequest searchRequest = new SearchRequest(INDEX_NAME);
        searchRequest.source(searchSourceBuilder);
        
        SearchHit[] searchHits = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT)
                .getHits()
                .getHits();
        
        return Arrays
                .stream(searchHits)
                .map(e -> Utils.toBean(e.getSourceAsString(), Position.class))
                .collect(Collectors.toList());
    }
    
    @PostMapping("/reindex")
    public Object reindex() throws IOException {
        
        CountDownLatch countDownLatch = new CountDownLatch(1);
    
        reCreateIndex(restHighLevelClient);
    
        // jdbcTemplate.setFetchSize(20);
        jdbcTemplate.query("select * from position", new RowCallbackHandler() {
        
            @Override
            public void processRow(ResultSet rs) throws SQLException {
            
                ResultSetMetaData resultSetMetaData = rs.getMetaData();
                int columnCount = resultSetMetaData.getColumnCount();
                
                List<Map<String, String>> data = new ArrayList<>();
            
                while (rs.next()) {
                
                    Map<String, String> line = new HashMap<>();
                
                    for (int i = 1; i <= columnCount; i++) {
                        line.put(resultSetMetaData.getColumnName(i), rs.getString(i));
                    }
                
                    data.add(line);
                }
                
                BulkProcessor bulkProcessor = getBulkProcessor(restHighLevelClient, countDownLatch);
                for (Map<String, String> datum : data) {
                    bulkProcessor.add(new IndexRequest(INDEX_NAME).source(datum));
                }
                bulkProcessor.flush();
            }
        });
        
        try {
            countDownLatch.await(1, TimeUnit.MINUTES);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        GetIndexRequest getIndexRequest = new GetIndexRequest(INDEX_NAME);
        GetIndexResponse getIndexResponse = restHighLevelClient.indices().get(getIndexRequest, RequestOptions.DEFAULT);
        
        return getIndexResponse.getIndices();
    }
    
    // 创建索引
    private void reCreateIndex(RestHighLevelClient client) throws IOException {
    
        IndicesClient indices = client.indices();
        
        // DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(INDEX_NAME);
        // indices.delete(deleteIndexRequest, RequestOptions.DEFAULT);
        
        // 创建索引mapping
        XContentBuilder builder = XContentFactory.jsonBuilder()
                .startObject()
                .field("properties")
                .startObject()
                
                .field("positionName")
                .startObject()
                .field("type", "text")
                .field("analyzer", "ik_max_word")
                .field("search_analyzer", "ik_smart")
                .endObject()
                
                .field("positionAdvantage")
                .startObject()
                .field("type", "text")
                .field("analyzer", "ik_max_word")
                .field("search_analyzer", "ik_smart")
                .endObject()
                
                // 定义其它字段
                
                .endObject()
                .endObject();
        
        CreateIndexRequest createIndexRequest = new CreateIndexRequest(INDEX_NAME);
        createIndexRequest.mapping(builder);
        
        CreateIndexResponse createIndexResponse = indices.create(createIndexRequest, RequestOptions.DEFAULT);
        // System.out.println(createIndexResponse.isAcknowledged());
    }
    
    
    private BulkProcessor getBulkProcessor(RestHighLevelClient client, CountDownLatch countDownLatch) {
        
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                System.out.println("beforeBulk");
            }
            
            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                countDownLatch.countDown();
                System.out.println("afterBulk");
            }
            
            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                countDownLatch.countDown();
                System.out.println("afterBulk  failure");
            }
        };
        
        BiConsumer<BulkRequest, ActionListener<BulkResponse>> bulkConsumer =
                (request, bulkListener) -> client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener);
        
        BulkProcessor.Builder builder = BulkProcessor.builder(bulkConsumer, listener);
        
        builder.setBulkActions(5000);
        builder.setBulkSize(new ByteSizeValue(100L, ByteSizeUnit.MB));
        builder.setConcurrentRequests(10);
        builder.setFlushInterval(TimeValue.timeValueSeconds(100L));
        builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3));
        
        return builder.build();
    }
}
